/*
分布式程序利用etcd来做配置中心，将一些配置信息放到 etcd 上进行集中管理。

这类场景的使用方式通常是这样：应用在启动的时候主动从 etcd 获取一次配置信息，同时，
在 etcd 节点上注册一个 Watcher 并等待，
以后每次配置有更新的时候，etcd 都会实时通知订阅者，以此达到获取最新配置信息的目的。

之前单机可以直接写个配置文件，程序初始化读取即可，但分布式程序每个都要单独配置太麻烦，修改还麻烦，
etcd就是在解决多台机器分布式配置问题。
*/
package mainfuncs

import (
	"context"
	"encoding/json"
	"fmt"
	"strings"
	"time"

	"LogManagementSystem/logagent/tailf"

	"github.com/astaxie/beego/logs"
	etcd_client "github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
)

var (
	etcdClient *EtcdClient
)

type EtcdClient struct {
	client *etcd_client.Client
	keys   []string
}

//初始化etcd,从etcd保存的配置信息中读取配置保存到collectConf，让tailf读取相关日志
func InitEtcd(addr string, key string) (collectConf []tailf.CollectConf, err error) {
	cli, err := etcd_client.New(etcd_client.Config{
		Endpoints:   []string{"localhost:2379", "localhost:22379", "localhost:32379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		logs.Error("connect etcd failed, err:", err)
		return
	}

	etcdClient = &EtcdClient{
		client: cli,
	}
	//判读是否是/结尾
	if strings.HasSuffix(key, "/") == false {
		key = key + "/"
	}

	//localIPArray数组存储全部后端ip，利用配置的key+ip作为每个后端独特辨识，作为完整etcdKey
	for _, ip := range localIPArray {
		etcdKey := fmt.Sprintf("%s%s", key, ip)
		etcdClient.keys = append(etcdClient.keys, etcdKey)

		ctx, cancel := context.WithTimeout(context.Background(), time.Second) //1s超时
		resp, err := cli.Get(ctx, etcdKey)                                    //读对应Key对应的value,也就是配置信息
		if err != nil {
			logs.Error("client get from etcd failed, err:%v", err)
			continue
		}
		cancel()

		logs.Debug("resp from etcd:%v", resp.Kvs)
		for _, v := range resp.Kvs {
			if string(v.Key) == etcdKey {
				err = json.Unmarshal(v.Value, &collectConf)
				if err != nil {
					logs.Error("unmarshal failed, err:%v", err)
					continue
				}

				logs.Debug("log config is %v", collectConf)
			}
		}
	}

	initEtcdWatcher()
	return
}

//对每个Key进行监听，是否有修改，key就是配置信息
func initEtcdWatcher() {

	for _, key := range etcdClient.keys {
		go watchKey(key)
	}
}

func watchKey(key string) {

	cli, err := etcd_client.New(etcd_client.Config{
		Endpoints:   []string{"localhost:2379", "localhost:22379", "localhost:32379"},
		DialTimeout: 5 * time.Second,
	})
	if err != nil {
		logs.Error("connect etcd failed, err:", err)
		return
	}

	logs.Debug("begin watch key:%s", key)
	for {
		rch := cli.Watch(context.Background(), key)
		var collectConf []tailf.CollectConf
		var getConfSucc = true

		for wresp := range rch { //管道读取
			for _, ev := range wresp.Events {

				//如果配置key被删除
				if ev.Type == mvccpb.DELETE {
					logs.Warn("key[%s] 's config deleted", key)
					continue
				}
				//如果配置key被修改
				if ev.Type == mvccpb.PUT && string(ev.Kv.Key) == key {
					err = json.Unmarshal(ev.Kv.Value, &collectConf)
					if err != nil {
						logs.Error("key [%s], Unmarshal[%s], err:%v ", err)
						getConfSucc = false
						continue
					}
				}
				logs.Debug("get config from etcd, %s %q : %q\n", ev.Type, ev.Kv.Key, ev.Kv.Value)
			}

			if getConfSucc {
				logs.Debug("get config from etcd succ, %v", collectConf)
				tailf.UpdateConfig(collectConf) //更新配置
			}
		}

	}
}
